package com.telra.app.server2.util;

import com.telra.app.server.cassandra.CassandraCQLQuery;
import com.telra.app.server.cassandra.CassandraService;
import com.telra.app.server.config.ConfigMgr;
import com.telra.app.server.data.OrderEvent;
import com.telra.app.server.protobuff.OrderProtos;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;

import java.util.List;
import java.util.concurrent.*;

/**
 * User: jamesarlet
 * Date: 10/7/12
 * Time: 10:45 AM
 */
public class ParallelThreadUtil {

    private void runTasks(final int n, final OrderEvent.Category category[]) {
        //init cassandra
        ApplicationContext ctx = new AnnotationConfigApplicationContext(ConfigMgr.class);
        CassandraService cassandraService = ctx.getBean(CassandraService.class);
        cassandraService.getKeyspace();

        //new way
        Long stime =  System.currentTimeMillis();
        ExecutorService threadPool = Executors.newFixedThreadPool(n);
        CompletionService<List<OrderProtos.OrderTicket>> pool =
                new ExecutorCompletionService<List<OrderProtos.OrderTicket>>(threadPool);

        for(int i = 0; i < n; i++){
            pool.submit(new QueryCass(category[i]));
        }

        for(int i = 0; i < n; i++){
            try {
                List<OrderProtos.OrderTicket> result = pool.take().get();
                System.err.println("result " + i + " has " + result.size() + " rows");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }

        threadPool.shutdown();
        Long etime =  System.currentTimeMillis();
        System.err.println("elapsed " + ((etime - stime)/1) );

        //old way
        stime =  System.currentTimeMillis();
        for (int i =0 ; i< 3; i++) {
            Callable<List<OrderProtos.OrderTicket>> c = new QueryCass(category[i]);
            try {
                List<OrderProtos.OrderTicket> orders = c.call();
                System.err.println("result " + i + " has " + orders.size() + " rows");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        etime =  System.currentTimeMillis();
        System.err.println("elapsed " + ((etime - stime)/1) );
    }

    private class QueryCass implements Callable<List<OrderProtos.OrderTicket>> {
        OrderEvent.Category category;
        public QueryCass(OrderEvent.Category category) {
            this.category = category;
        }
        @Override
        public List<OrderProtos.OrderTicket> call() throws Exception {
            CassandraCQLQuery query = CassandraCQLQuery.getInstance();
            return query.queryByCategory(category, 5000);
        }
    }

    public static void main(String[] args) {
        new ParallelThreadUtil().runTasks(3,
                new OrderEvent.Category[]{OrderEvent.Category.FOOTWEAR,
                OrderEvent.Category.ACCESSORIES,OrderEvent.Category.OUTERWEAR});
    }

}